package io.github.wanggit.antrpc.console.zookeeper;

import io.github.wanggit.antrpc.commons.bean.*;
import io.github.wanggit.antrpc.commons.constants.ConstantValues;
import io.github.wanggit.antrpc.commons.utils.ConfigPathUtil;
import io.github.wanggit.antrpc.commons.utils.JsonUtils;
import io.github.wanggit.antrpc.commons.zookeeper.NodeUnregDTO;
import io.github.wanggit.antrpc.commons.zookeeper.ZkNodeType;
import io.github.wanggit.antrpc.console.domain.ConsumerRateLimiting;
import io.github.wanggit.antrpc.console.domain.ProviderRateLimiting;
import io.github.wanggit.antrpc.console.service.ConsumerRateLimitingService;
import io.github.wanggit.antrpc.console.service.ProviderRateLimitingService;
import io.github.wanggit.antrpc.console.web.utils.LongToDateUtil;
import io.github.wanggit.antrpc.console.web.vo.ConfigVO;
import io.github.wanggit.antrpc.console.web.vo.SubscribeConfigVO;
import io.github.wanggit.antrpc.console.web.vo.ZookeeperNodeVO;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.utils.ZKPaths;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.Charset;
import java.util.List;
import java.util.Map;
import java.util.Objects;

@Slf4j
@Component
public class ZookeeperListener {

    @Value("${antrpc.zk.node.prefixs}")
    private String nodePrefixs;

    @Autowired private IInterfaceContainer interfaceContainer;

    @Autowired private ISubscribeContainer subscribeContainer;

    @Autowired private IAllNodeContainer allNodeContainer;

    @Autowired private IUnregNodeContainer unregNodeContainer;

    @Autowired private IConfigContainer configContainer;

    @Autowired private ISubscribeConfigContainer subscribeConfigContainer;

    @Autowired private IZkClient zkClient;

    @Autowired private IZkNodeOperator zkNodeOperator;

    @Autowired private ConsumerRateLimitingService consumerRateLimitingService;

    @Autowired private ProviderRateLimitingService providerRateLimitingService;

    private CuratorFramework curatorFramework;

    private String[] nodePrefixArr;

    @PostConstruct
    public void init() throws Exception {
        nodePrefixArr = null == nodePrefixs ? new String[] {} : nodePrefixs.split("\\s*,\\s*");
        curatorFramework = zkClient.getCuratorFramework();
        initProviderTreeCacheListener();
        initSubscribeTreeCacheListener();
        initAllNodeTreeCacheListener();
        initUnregNodeTreeCacheListener();
        initConfigTreeCacheListener();
        initSubscribeConfigTreeCacheListener();
    }

    /**
     * 监听所有订阅配置的节点变化
     *
     * @throws Exception e
     */
    private void initSubscribeConfigTreeCacheListener() throws Exception {
        TreeCache treeCache =
                new TreeCache(curatorFramework, "/" + ConstantValues.ZK_ROOT_SUBSCRIBE_CONFIG_NAME);
        treeCache
                .getListenable()
                .addListener(
                        (client, event) -> {
                            if (Objects.equals(event.getType(), TreeCacheEvent.Type.NODE_ADDED)
                                    || Objects.equals(
                                            event.getType(), TreeCacheEvent.Type.NODE_UPDATED)) {
                                ChildData childData = event.getData();
                                if (null != childData) {
                                    // /__rpc_subcnf__/env/spaceName/hostInfo
                                    String path = childData.getPath();
                                    if (Objects.equals(
                                            ZkNodeType.SubscribeConfigNodeType.KEY,
                                            ZkNodeType.getSubscribeConfigNodeType(path))) {
                                        SubscribeConfigVO subscribeConfigVO =
                                                createSubscribeConfigVO(childData, path);
                                        subscribeConfigContainer.saveOrUpdate(subscribeConfigVO);
                                    }
                                }
                            } else if (Objects.equals(
                                    event.getType(), TreeCacheEvent.Type.NODE_REMOVED)) {
                                ChildData childData = event.getData();
                                if (null != childData) {
                                    // /__rpc_subcnf__/env/spaceName/hostInfo
                                    String path = childData.getPath();
                                    if (Objects.equals(
                                            ZkNodeType.SubscribeConfigNodeType.KEY,
                                            ZkNodeType.getSubscribeConfigNodeType(path))) {
                                        SubscribeConfigVO subscribeConfigVO =
                                                createSubscribeConfigVO(childData, path);
                                        subscribeConfigContainer.delete(subscribeConfigVO);
                                    }
                                }
                            }
                        });
        treeCache.start();
    }

    private SubscribeConfigVO createSubscribeConfigVO(ChildData childData, String path) {
        List<String> items = ZKPaths.split(path);
        SubscribeConfigVO subscribeConfigVO = new SubscribeConfigVO();
        subscribeConfigVO.setFullPath(path);
        subscribeConfigVO.setEnvName(items.get(1));
        subscribeConfigVO.setSpaceName(items.get(2));
        subscribeConfigVO.setHostInfo(items.get(3));
        SubscribeConfigData subscribeConfigData =
                JsonUtils.fromJson(childData.getData(), SubscribeConfigData.class);
        subscribeConfigVO.setTs(LongToDateUtil.toDate(subscribeConfigData.getTs()));
        return subscribeConfigVO;
    }

    /**
     * 监听所有配置节点
     *
     * @throws Exception e
     */
    private void initConfigTreeCacheListener() throws Exception {
        TreeCache treeCache =
                new TreeCache(curatorFramework, "/" + ConstantValues.ZK_ROOT_CONFIG_NAME);
        treeCache
                .getListenable()
                .addListener(
                        (client, event) -> {
                            if (Objects.equals(event.getType(), TreeCacheEvent.Type.NODE_ADDED)
                                    || Objects.equals(
                                            event.getType(), TreeCacheEvent.Type.NODE_UPDATED)) {
                                ChildData childData = event.getData();
                                if (null != childData) {
                                    String path = childData.getPath();
                                    if (Objects.equals(
                                            ZkNodeType.getConfigNodeType(path),
                                            ZkNodeType.ConfigNodeType.ENV)) {
                                        // /__rpc_cnf__/dev
                                        ConfigVO configVO = createEnvConfigVO(path);
                                        configContainer.saveOrUpdateEnv(configVO);
                                    } else if (Objects.equals(
                                            ZkNodeType.getConfigNodeType(path),
                                            ZkNodeType.ConfigNodeType.SPACE)) {
                                        // /__rpc_cnf__/dev/service_a
                                        ConfigVO configVO = createSpaceConfigVO(path);
                                        configContainer.saveOrUpdateSpace(configVO);
                                    } else if (Objects.equals(
                                            ZkNodeType.getConfigNodeType(path),
                                            ZkNodeType.ConfigNodeType.KEY)) {
                                        // /__rpc_cnf__/dev/service_a/config_name
                                        ConfigVO configVO = dataToConfigVO(childData, path);
                                        configContainer.saveOrUpdate(configVO);
                                    }
                                }
                            } else if (Objects.equals(
                                    event.getType(), TreeCacheEvent.Type.NODE_REMOVED)) {
                                ChildData childData = event.getData();
                                if (null != childData) {
                                    String path = childData.getPath();
                                    if (Objects.equals(
                                            ZkNodeType.getConfigNodeType(path),
                                            ZkNodeType.ConfigNodeType.ENV)) {
                                        // /__rpc_cnf__/dev
                                        ConfigVO configVO = createEnvConfigVO(path);
                                        configContainer.deleteEnv(configVO);
                                    } else if (Objects.equals(
                                            ZkNodeType.getConfigNodeType(path),
                                            ZkNodeType.ConfigNodeType.SPACE)) {
                                        // /__rpc_cnf__/dev/service_a
                                        ConfigVO configVO = createSpaceConfigVO(path);
                                        configContainer.deleteSpace(configVO);
                                    } else if (Objects.equals(
                                            ZkNodeType.getConfigNodeType(path),
                                            ZkNodeType.ConfigNodeType.KEY)) {
                                        ConfigVO configVO = dataToConfigVO(childData, path);
                                        configContainer.delete(configVO);
                                    }
                                    // delete /__rpc_subcnf__/***
                                    deleteSubscribeConfigNode(path);
                                }
                            }
                        });
        treeCache.start();
    }

    public void deleteSubscribeConfigNode(String configPath) {
        String subscribePath = ConfigPathUtil.castPathToSubscribePath(configPath);
        try {
            zkNodeOperator.deleteNode(subscribePath);
        } catch (Exception e) {
            if (log.isWarnEnabled()) {
                log.warn("delete " + subscribePath + " error. " + e.getMessage());
            }
        }
    }

    private ConfigVO createSpaceConfigVO(String path) {
        List<String> items = ZKPaths.split(path);
        ConfigVO configVO = new ConfigVO();
        configVO.setEnv(items.get(1));
        configVO.setSpaceName(items.get(2));
        configVO.setFullPath(path);
        return configVO;
    }

    private ConfigVO createEnvConfigVO(String path) {
        List<String> items = ZKPaths.split(path);
        ConfigVO configVO = new ConfigVO();
        configVO.setEnv(items.get(1));
        configVO.setFullPath(path);
        return configVO;
    }

    private ConfigVO dataToConfigVO(ChildData childData, String path) {
        List<String> names = ZKPaths.split(path);
        ConfigItemData configItemData = ConfigItemData.from(childData.getData());
        ConfigVO configVO = new ConfigVO();
        configVO.setConfigItemData(configItemData);
        configVO.setEnv(names.get(1));
        configVO.setSpaceName(names.get(2));
        configVO.setFullPath(path);
        return configVO;
    }

    /**
     * 监听所有临时下线的节点
     *
     * @throws Exception e
     */
    private void initUnregNodeTreeCacheListener() throws Exception {
        TreeCache treeCache =
                new TreeCache(curatorFramework, "/" + ConstantValues.ZK_ROOT_NODE_UNREG_NAME);
        treeCache
                .getListenable()
                .addListener(
                        (client, event) -> {
                            if (TreeCacheEvent.Type.NODE_ADDED.equals(event.getType())) {
                                NodeUnregDTO nodeUnregDTO = getNodeUnregDTOFromEvent(event);
                                if (null != nodeUnregDTO) {
                                    unregNodeContainer.add(nodeUnregDTO);
                                }
                            } else if (TreeCacheEvent.Type.NODE_REMOVED.equals(event.getType())) {
                                NodeUnregDTO nodeUnregDTO = getNodeUnregDTOFromEvent(event);
                                if (null != nodeUnregDTO) {
                                    unregNodeContainer.delete(nodeUnregDTO);
                                }
                            }
                        });
        treeCache.start();
    }

    private NodeUnregDTO getNodeUnregDTOFromEvent(TreeCacheEvent event) {
        ChildData childData = event.getData();
        if (null != childData) {
            String path = childData.getPath();
            ZkNodeType.UnregNodeType type = ZkNodeType.getUnregNodeType(path);
            if (ZkNodeType.UnregNodeType.INTERFACE.equals(type)) {
                return JsonUtils.fromJson(childData.getData(), NodeUnregDTO.class);
            }
        }
        return null;
    }

    /**
     * 为显示zookeeper节点而进行的所有节点监听
     *
     * @throws Exception e
     */
    private void initAllNodeTreeCacheListener() throws Exception {
        TreeCache treeCache = new TreeCache(curatorFramework, "/");
        treeCache
                .getListenable()
                .addListener(
                        (client, event) -> {
                            if (TreeCacheEvent.Type.NODE_ADDED.equals(event.getType())) {
                                ZookeeperNodeVO zookeeperNodeVO = createZookeeperNodeVO(event);
                                if (null != zookeeperNodeVO) {
                                    allNodeContainer.add(zookeeperNodeVO);
                                }
                            } else if (TreeCacheEvent.Type.NODE_UPDATED.equals(event.getType())) {
                                ZookeeperNodeVO zookeeperNodeVO = createZookeeperNodeVO(event);
                                if (null != zookeeperNodeVO) {
                                    allNodeContainer.update(zookeeperNodeVO);
                                }
                            } else if (TreeCacheEvent.Type.NODE_REMOVED.equals(event.getType())) {
                                ZookeeperNodeVO zookeeperNodeVO = createZookeeperNodeVO(event);
                                if (null != zookeeperNodeVO) {
                                    allNodeContainer.delete(zookeeperNodeVO);
                                }
                            }
                        });
        treeCache.start();
    }

    private ZookeeperNodeVO createZookeeperNodeVO(TreeCacheEvent event) {
        ChildData childData = event.getData();
        if (null == childData) {
            return null;
        }
        String path = childData.getPath();
        for (String prefix : nodePrefixArr) {
            if (path.startsWith(prefix.trim())) {
                ZookeeperNodeVO vo = new ZookeeperNodeVO();
                vo.setPath(path);
                vo.setData(new String(childData.getData(), Charset.forName("UTF-8")));
                return vo;
            }
        }
        return null;
    }

    /**
     * 监听服务消费者
     *
     * @throws Exception
     */
    private void initSubscribeTreeCacheListener() throws Exception {
        TreeCache subscribeRootNodeTreeCache =
                new TreeCache(curatorFramework, "/" + ConstantValues.ZK_ROOT_SUBSCRIBE_NODE_NAME);
        subscribeRootNodeTreeCache
                .getListenable()
                .addListener(
                        (client, event) -> {
                            if (TreeCacheEvent.Type.NODE_ADDED.equals(event.getType())
                                    || TreeCacheEvent.Type.NODE_UPDATED.equals(event.getType())
                                    || TreeCacheEvent.Type.NODE_REMOVED.equals(event.getType())) {
                                doSubscribeEventOnType(event);
                            }
                        });
        subscribeRootNodeTreeCache.start();
    }

    /**
     * 监听服务提供者
     *
     * @throws Exception
     */
    private void initProviderTreeCacheListener() throws Exception {
        TreeCache providerRootNodeTreeCache =
                new TreeCache(curatorFramework, "/" + ConstantValues.ZK_ROOT_NODE_NAME);
        providerRootNodeTreeCache
                .getListenable()
                .addListener(
                        (client, event) -> {
                            if (TreeCacheEvent.Type.NODE_ADDED.equals(event.getType())
                                    || TreeCacheEvent.Type.NODE_UPDATED.equals(event.getType())
                                    || TreeCacheEvent.Type.NODE_REMOVED.equals(event.getType())) {
                                doProviderEventOnType(event);
                            }
                        });
        providerRootNodeTreeCache.start();
    }

    private void doSubscribeEventOnType(TreeCacheEvent event) throws Exception {
        ChildData childData = event.getData();
        ZkNodeType.RegisterNodeType type = ZkNodeType.getRegisterNodeType(childData.getPath());
        if (ZkNodeType.RegisterNodeType.INTERFACE.equals(type)) {
            SubscribeNode subscribeNode =
                    JsonUtils.fromJson(childData.getData(), SubscribeNode.class);
            if (TreeCacheEvent.Type.NODE_ADDED.equals(event.getType())) {
                subscribeContainer.addSubscribeNode(subscribeNode);
                // 因为节点类型为临时节点，所以在新建时可能会丢失以前设置的限流信息，所以在此要进行限流信息恢复
                ConsumerRateLimiting consumerRateLimiting = new ConsumerRateLimiting();
                consumerRateLimiting.setAppNameHostPort(subscribeNode.getHost());
                consumerRateLimiting.setClassName(subscribeNode.getClassName());
                Map<String, RegisterBeanMethod> registerBeanMethodMap =
                        consumerRateLimitingService.findConsumerRateLimitingByClass(
                                consumerRateLimiting);
                if (!registerBeanMethodMap.isEmpty()) {
                    // 直接更新到远程的Zookeeper节点
                    subscribeNode.setMethodMap(registerBeanMethodMap);
                    zkNodeOperator.setNodeData(
                            childData.getPath(), JsonUtils.toJsonBytes(subscribeNode));
                }
            } else if (TreeCacheEvent.Type.NODE_UPDATED.equals(event.getType())) {
                subscribeContainer.addSubscribeNode(subscribeNode);
            } else if (TreeCacheEvent.Type.NODE_REMOVED.equals(event.getType())) {
                subscribeContainer.deleteSubscribeNode(subscribeNode);
            }
        }
    }

    private void doProviderEventOnType(TreeCacheEvent event) throws Exception {
        ChildData childData = event.getData();
        ZkNodeType.RegisterNodeType type = ZkNodeType.getRegisterNodeType(childData.getPath());
        if (ZkNodeType.RegisterNodeType.INTERFACE.equals(type)) {
            InterfaceNodeDataBean interfaceNodeDataBean =
                    JsonUtils.fromJson(childData.getData(), InterfaceNodeDataBean.class);
            if (TreeCacheEvent.Type.NODE_ADDED.equals(event.getType())) {
                interfaceContainer.addInterface(interfaceNodeDataBean);
                // 检查是否有入库的服务提供端限流信息，如果有就恢复
                ProviderRateLimiting providerRateLimiting = new ProviderRateLimiting();
                providerRateLimiting.setHostPort(interfaceNodeDataBean.getHostInfo());
                providerRateLimiting.setClassName(interfaceNodeDataBean.getClassName());
                Map<String, RegisterBeanMethod> registerBeanMethodMap =
                        providerRateLimitingService.findProviderRateLimiting(providerRateLimiting);
                if (!registerBeanMethodMap.isEmpty()) {
                    registerBeanMethodMap.forEach(
                            (key, value) -> interfaceNodeDataBean.getMethodMap().put(key, value));
                    // 设置恢复限流之后的数据，再次触发节点数据更新
                    zkNodeOperator.setNodeData(
                            childData.getPath(), JsonUtils.toJsonBytes(interfaceNodeDataBean));
                }
            } else if (TreeCacheEvent.Type.NODE_UPDATED.equals(event.getType())) {
                interfaceContainer.updateInterface(interfaceNodeDataBean);
            } else if (TreeCacheEvent.Type.NODE_REMOVED.equals(event.getType())) {
                interfaceContainer.deleteInterface(interfaceNodeDataBean);
            }
        }
    }
}
